package com.runjian.rundo.iot.agent.business.kafka;

import java.util.Optional;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;

import com.alibaba.fastjson2.JSON;
import com.runjian.rundo.bmp.kernel.common.spring.context.utils.MdcUtils;
import com.runjian.rundo.iot.agent.business.control.executor.DeviceCommandExecutor;
import com.runjian.rundo.iot.agent.common.entity.CommonDeviceCommand;

import cn.hutool.core.util.IdUtil;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * IoT 设备控制 mq处理
 *
 * @author LiGang
 */
@RequiredArgsConstructor
@Slf4j
public class KafkaControlConsumer {

    private final DeviceCommandExecutor deviceCommandExecutor;

    @KafkaListener(topics = {"#{'${spring.kafka.topics.controlDown}'.split(',')}"})
    public void defaultListen(ConsumerRecord<String, String> record) {

        Optional.ofNullable(record.value()).ifPresent(message -> {
            // 重新生成链路追踪信息，防止线程复用产生的链路追踪不准确
            MdcUtils.setTraceIdIfAbsent(IdUtil.simpleUUID());
            if (log.isTraceEnabled()) {
                log.trace("【KafKa控制消费记录】:{}", record);
            }
            try {
                CommonDeviceCommand commonDeviceCommand = JSON.parseObject(message, CommonDeviceCommand.class);
                MdcUtils.setTraceIdIfAbsent(commonDeviceCommand.getTraceId());
                // 控制执行
                deviceCommandExecutor.execute(commonDeviceCommand);
            } catch (Exception e) {
                log.error("KafKa控制消费异常:{},error:", message, e);
            }
        });

    }


}
